# coding=utf-8
import configparser
import json
import os
import sys
import time

from kafka import KafkaConsumer, TopicPartition
from pip._vendor import requests


def configInfo():
    global broker_list, topic_list, lag_max
    broker_list = []
    topic_list = []
    configBroker = configparser.ConfigParser()
    configTopic = configparser.ConfigParser()
    print('正在读取conf配置')
    configBroker.read('./conf/brokers.conf', encoding='utf-8')
    configTopic.read('./conf/topics.conf', encoding='utf-8')
    # print('sections:', ' ', config.sections())
    # 获取broker配置
    broker_list = configBroker.get('config', 'broker').split(",")
    # 获取topic配置
    topic_list = configTopic.get('config', 'topic').split(",")
    # 阈值设设置
    lag_max = configTopic.get('config', 'lag_max')


def calLogic():
    all_lags = []
    # TODO 优化 topic比较多 多个进程去跑 或者是多线程
    for topic in topic_list:
        currentTopic = topic
        # print("###"+currentTopic)
        if "tendency" in currentTopic:
            group_id = topic + '_kafka2db'
        # 非情感计算
        elif "v2" in currentTopic:
            group_id = topic + '_kafka2db'
        else:
            group_id = currentTopic + '_tendency2kfk'
        consumer = KafkaConsumer(topic,
                                 group_id=group_id,
                                 bootstrap_servers=broker_list)
        print(consumer.partitions_for_topic(topic))
        # partitions = [TopicPartition(topic, p ) for p in consumer.partitions_for_topic(topic)]
        partitions = []
        for_topic = consumer.partitions_for_topic(topic)
        if for_topic is not None:
            for p in consumer.partitions_for_topic(topic):
                if p is not None:
                    partitions.append(TopicPartition(topic, p))
        print("start to cal offset:")

        # total
        toff = consumer.end_offsets(partitions)
        toff = [(key.partition, toff[key]) for key in toff.keys()]
        toff.sort()
        print("total offset: {}".format(str(toff)))

        # current
        coff = [(x.partition, consumer.committed(x)) for x in partitions]
        coff.sort()
        print("current offset: {}".format(str(coff)))

        # cal sum and left
        toff_sum = sum([x[1] for x in toff])
        cur_sum = sum([x[1] for x in coff if x[1] is not None])
        left_sum = toff_sum - cur_sum
        print("topic:" + currentTopic + ",kafka left: {}".format(left_sum))

        if int(lag_max) <= int(left_sum):
            all_lags.append((topic, left_sum))
    return all_lags


def createLogFile():
    # ----获取时间----创建文件夹----
    tt = time.strftime("%Y%m%d---%H%M", time.localtime())
    path = r"/Users/lxp/PyCharmProjects/TestKafkaMonitor/logs"
    if not os.path.exists(path):
        os.makedirs(path)
        print('文件夹创建完成  ' + path)


def writeToFile(file):
    # print("file:" + file)
    file_handle = open('./logs/log.txt', mode='a')
    file_handle.write(file + '\n')
    file_handle.close()


def sendEmall(all_lags):
    app_id = 'JJtfwsYi'
    app_secret = '902NCUKm8j3yn7JX'
    access_token = getAccessToken(app_id, app_secret)
    # 初始化邮件json格式
    jsonData = {"msgtype": "email", "msgtag": "iaas", "enable_custom_notice": 0, "custom_notice": [],
                "body": {"ctype": "html", "subject": "TestTT", "from_alias": "LLXX",
                         "to": ["919055887@qq.com", "lixinpeng@mininglamp.com"], "content": "测试邮件",
                         "attachments": ["cee70de0bfed7dfcabaa7aa40b04c1f9Jz17X0j(",
                                         "e709ce11ab4541fdd8aa3f8c7d3edb664pscaC6*"]}}
    # 主题
    # subject = "<font size='8' color='black''> 【TOPIC监控】</font>"
    subject = "【Topic 监控】"
    # 发件人from_alias
    fromAlias = 'social@admaster.com.cn'
    # 收件人
    toMembers = ["919055887@qq.com", "lixinpeng@mininglamp.com"]
    # 发送内容
    # sendCcontent = "你好，<p>这是使用python自动发送的HTML格式的测试邮件：</p> <p> <a href='https://www.python.org'> {all_lags}</a></p>".format(
    #     all_lags=all_lags.rstrip())
    sendCcontent = all_lags
    jsonData['body']['subject'] = subject
    jsonData['body']['from_alias'] = fromAlias
    jsonData['body']['to'] = toMembers
    jsonData['body']['content'] = sendCcontent
    # 发送邮件
    postEmall(access_token, jsonData)


def postEmall(access_token, json_data):
    pUrl = "https://account.mlamp.cn/api/msg/send/?access_token={access_token}".format(
        access_token=access_token.rstrip())
    r1 = requests.post(pUrl, json=json_data)


def getAccessToken(app_id, app_secret):
    getUrl = "https://account.mlamp.cn/api/token/gettoken/?app_id={app_id}&app_secret={app_secret}".format(
        app_id=app_id.rstrip(), app_secret=app_secret.rstrip())
    r0 = requests.get(getUrl)
    print(r0.content)
    data2 = json.loads(r0.content)
    print("data2['access_token']: ", data2['access_token'])
    access_token = data2['access_token']
    return access_token


# 发送邮件
# TODO 获取生成速率和消费速度
# # 判断 两个速率和 lag的算法
if __name__ == '__main__':
    # while True:
        print("start:"+time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
        configInfo()
        all_lags = calLogic()
        if len(all_lags) > 0:
            # sys.exit(0)
            print("No topic exceeds the warning value ")
            allContent = "<font size='5'>&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp报警 <br /> </font>"
            for lags in all_lags:
                # sendCcontent = "你好，<p>这是使用python自动发送的HTML格式的测试邮件：</p> <p> <a href='https://www.python.org'> {all_lags}</a></p>".format(
                allContent += " TOPIC &nbsp:<font size='5'>&nbsp{} <br/> </font>" \
                              "LEFT: <font size='5' color='red'>{}<br/><br/> </font>\r\n\r\n".format(lags[0], lags[1])
                # writeToFile("***报警***\n topic:{},group_id:{},left:{}\n"
                #             .format(lags[0], lags[1], lags[2]))
            sendEmall(allContent)
        else:
            print("No topic exceeds the warning value ")
        print("end:" + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
        # time.sleep(3600 / 2)
